package com.example.commons;

import com.example.commons.future.MsgFuture;
import com.example.commons.future.SyncMsgFuture;
import com.example.commons.model.BizReqMsg;
import com.example.commons.model.BizResMsg;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
@Component
public class MsgPool {

    /**
     * 用于记录请求的消息池
     */
    private final static Map<String, MsgFuture> MSG_POOL = new ConcurrentHashMap<>();

    /**
     * 用于记录客户端链接池
     */
    private final static Map<String, Channel> CLIENT = new ConcurrentHashMap<>();

    public final static String DEFAULT_CMD = "*";

    /**
     * 用于记录服务端的链接通道
     */
    private Channel serverChannel;

    private Channel getClient(String clientId) {
        return CLIENT.get(clientId);
    }

    public void putClient(String clientId, Channel channel) {
        log.info("注册客户端连接[{}]", clientId);
        synchronized (clientId) {
            CLIENT.put(clientId, channel);
        }
    }

    public void removeClient(String clientId) {
        log.info("注销客户端连接[{}]", clientId);
        synchronized (clientId) {
            CLIENT.remove(clientId);
        }
    }

    public <REQ, RES> RES sendMsgToServer(REQ req) {
        return sendMsg(serverChannel, DEFAULT_CMD, req, 5000);
    }

    public <REQ, RES> RES sendMsgToServer(String cmd, REQ req) {
        if (serverChannel == null) {
            throw new NullPointerException("serverChannel");
        }
        return sendMsg(serverChannel, cmd, req, 5000);
    }

    public <REQ, RES> RES sendMsgToClient(String clientId, REQ req) {
        return sendMsgToClient(clientId, DEFAULT_CMD, req);
    }

    public <REQ, RES> RES sendMsgToClient(String clientId, String cmd, REQ req) {
        Channel channel = getClient(clientId);
        if (channel != null) {
            return sendMsg(channel, cmd, req, 5000);
        } else {
            throw new RuntimeException("前置程序未配置");
        }
    }

    public void sendMsg(Channel channel, Object msg) {
        if (channel == null) {
            throw new NullPointerException("channel");
        }
        if (msg == null) {
            throw new NullPointerException("request");
        }

        channel.writeAndFlush(msg);
    }

    /**
     * 发送同步消息并获取返回结果
     *
     * @param channel
     * @param req
     * @param timeout
     * @param req
     * @return
     */
    public <REQ, RES> RES sendMsg(final Channel channel, String cmd, final REQ req, final long timeout) {
        if (channel == null) {
            throw new NullPointerException("channel");
        }
        if (req == null) {
            throw new NullPointerException("request");
        }
        if (timeout <= 0) {
            throw new IllegalArgumentException("timeout <= 0");
        }
        BizReqMsg<REQ> reqMsg = new BizReqMsg<>();
        reqMsg.setCmd(cmd);
        reqMsg.setData(req);

        if (reqMsg.getReqId() == null) {
            String reqId = UUID.randomUUID().toString().replaceAll("-", "");
            reqMsg.setReqId(reqId);
        }

        MsgFuture<RES> future = new SyncMsgFuture<>(reqMsg.getReqId());
        MSG_POOL.put(reqMsg.getReqId(), future);

        BizResMsg<RES> response;
        try {
            response = doReqAndRes(channel, reqMsg, timeout, future);
        } catch (Exception e) {
            log.error("netty通讯异常", e);
            throw new RuntimeException(e.getMessage());
        }

        MSG_POOL.remove(reqMsg.getReqId());
        return response.getResult();
    }

    /**
     * 同步获取返回结果
     *
     * @param channel
     * @param req
     * @param timeout
     * @param msgFuture
     * @return
     * @throws Exception
     */
    private <REQ, RES> BizResMsg<RES> doReqAndRes(final Channel channel,
                                                  final BizReqMsg<REQ> req,
                                                  final long timeout,
                                                  final MsgFuture<RES> msgFuture) throws Exception {
        channel.writeAndFlush(req).addListener((ChannelFutureListener) future -> {
            msgFuture.setWriteResult(future.isSuccess());
            msgFuture.setCause(future.cause());
            //失败移除
            if (!msgFuture.isWriteSuccess()) {
                MSG_POOL.remove(msgFuture.setReqId());
            }
        });

        BizResMsg<RES> response = msgFuture.get(timeout, TimeUnit.MILLISECONDS);
        if (response == null) {
            if (msgFuture.isTimeout()) {
                throw new TimeoutException(String.format("通讯超时 begin:[%s] [%s]", msgFuture.getBegin(), timeout));
            } else {
                // write exception
                throw new Exception(msgFuture.cause());
            }
        }
        return response;
    }

    public <RES> MsgFuture<RES> get(String reqId) {
        return MSG_POOL.get(reqId);
    }

    public synchronized void setServerChannel(Channel channel) {
        this.serverChannel = channel;
    }

    public Channel getServerChannel() {
        return serverChannel;
    }
}
